package com.ibeifeng.bigdata.senior.hadoop.mapreduce;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class WCMapReduce extends Configured implements Tool{

	// step 1: Mapper Class
	/**
	 * public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
	 */
	public static class WCMapper extends
			Mapper<LongWritable, Text, Text, IntWritable> {
		private Text mapOutputKey = new Text();
		private final static IntWritable mapOutputValue = new IntWritable(1);
		
		/**
		 * protected void map(KEYIN key, VALUEIN value, Context context)
		 */
		@Override
		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			System.out.println("map-in-key:" + key.get() + " -- "//
					+ "map-in-value:" + value.toString());
			
			
			// line value
			String lineValue = value.toString();
			
			// split
			String[] strs = lineValue.split(" ") ;
			
			// iterator
			for(String str: strs){
				// set map output key
				mapOutputKey.set(str);
				
				// output
				context.write(mapOutputKey, mapOutputValue);
			}
		}

	}

	// step 2: Reduce Class
	/**
	 * public class Reducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT>
	 */
	public static class WCReducer extends
			Reducer<Text, IntWritable, Text, IntWritable> {
		private IntWritable outputValue = new IntWritable();

		/**
		 *  protected void reduce(KEYIN key, Iterable<VALUEIN> values, Context context )
		 */
		@Override
		public void reduce(Text key, Iterable<IntWritable> values,
				Context context) throws IOException, InterruptedException {
			System.out.println("red-in-key: " + key.toString() + " -- "//
				+ "red-in-value: " + values);
			
			// temp : sum
			int sum = 0 ;
			
			// iterator
			for(IntWritable value : values){
				// total
				sum += value.get() ;
			}
			
			// set output value 
			outputValue.set(sum);
			
			// output
			context.write(key, outputValue);
		}

	}

	// step 3: Driver
	/**
	 * int run(String [] args) throws Exception;
	 */
	public int run(String[] args) throws Exception {
		// 1: get Configuration
		Configuration configuration = this.getConf();

		// 2: create job
		Job job = Job.getInstance(configuration,//
			this.getClass().getSimpleName()//
		);
		job.setJarByClass(this.getClass());
		
		// 3: set job
		// input  -> map  -> reduce -> output
		// 3.1: input
		Path inPath = new Path(args[0]);
		FileInputFormat.addInputPath(job, inPath);
		
		// 3.2: mapper
		job.setMapperClass(WCMapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		
		// 3.3: reducer
		job.setReducerClass(WCReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		// 3.4: output
		Path outPath = new Path(args[1]);
		FileOutputFormat.setOutputPath(job, outPath);
		
		// 4: submit job
		boolean isSuccess = job.waitForCompletion(true) ;
		
		return isSuccess ? 0 : 1 ;
	}
	
	
	public static void main(String[] args) throws Exception {
		/**
		args = new String[]{ //
			"hdfs://bigdata-senior01.ibeifeng.com:8020/user/beifeng/input" ,//
			"hdfs://bigdata-senior01.ibeifeng.com:8020/user/beifeng/output5"
		};
		*/

		// get configuration
		Configuration configuration = new Configuration();
		
		// run job
		int status  = ToolRunner.run(// 
			configuration,//
			new WCMapReduce(),//
			args //
		) ;
		
		//exit program
		System.exit(status);
	}


}
